package learn

import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala.{OutputTag, StreamExecutionEnvironment}
import org.apache.flink.util.Collector

object test01SideOutput {
    def main(args: Array[String]): Unit = {
        val tag = new OutputTag[(String, String)]("x")
        //val tag2 = new OutputTag[String]("x")

        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)

        val stream = env.socketTextStream("hadoop000", 9527).process(new ProcessFunction[String, String] {
            override def processElement(value: String, ctx: ProcessFunction[String, String]#Context, out: Collector[String]): Unit = {
                if (value == "pk"){
                    ctx.output(new OutputTag[(String, String)]("x"), (value, value))
                } else {
                    out.collect(value)
                }
            }
        })
        stream.print("---ok---")
        stream.getSideOutput(tag).print("---side---")

        env.execute("stageTest")
    }
}
